package com.atguigu.gmall.realtime.util;

import com.atguigu.gmall.realtime.commont.Constant;

/**
 * @Author lzc
 * @Date 2023/4/23 10:11
 */
public class SQLUtil {
    
    public static String getKafkaSourceDDL(String topic, String groupId, String... format) {
        String defaultFormat = "json";
        if (format.length > 0) {
            defaultFormat = format[0];
        }
        return "with(" +
            "  'connector' = 'kafka', " +
            "  'topic' = '" + topic + "', " +
            "  'properties.bootstrap.servers' = '" + Constant.KAFKA_SERVERS + "', " +
            "  'properties.group.id' = '" + groupId + "', " +
            "  'scan.startup.mode' = 'latest-offset', " +
            ("json".equals(defaultFormat) ? " 'json.ignore-parse-errors' = 'true', " : "") +
            "  'format' = '" + defaultFormat + "' " +
            ")";
    }
    
    public static String getKafkaSinkDDL(String topic, String... format) {
        String defaultFormat = "json";
        if (format.length > 0) {
            defaultFormat = format[0];
        }
        return "with(" +
            "  'connector' = 'kafka', " +
            "  'topic' = '" + topic + "', " +
            "  'properties.bootstrap.servers' = '" + Constant.KAFKA_SERVERS + "', " +
            ("json".equals(defaultFormat) ? " 'json.ignore-parse-errors' = 'true', " : "") +
            "  'format' = '" + defaultFormat + "' " +
            ")";
    }
    
    public static String getUpsertKafkaDDL(String topic, String ... format) {
        String defaultFormat = "json";
        if (format.length > 0) {
            defaultFormat = format[0];
        }
        return "with(" +
            "  'connector' = 'upsert-kafka', " +
            "  'topic' = '" + topic + "', " +
            "  'properties.bootstrap.servers' = '" + Constant.KAFKA_SERVERS + "', " +
            ("json".equals(defaultFormat) ? " 'key.json.ignore-parse-errors' = 'true', " : "") +
            ("json".equals(defaultFormat) ? " 'value.json.ignore-parse-errors' = 'true', " : "") +
            "  'key.format' = '" + defaultFormat + "',  " +
            "  'value.format' = '" + defaultFormat + "'  " +
            ")";
    }
}
/*
 "  'json.fail-on-missing-field' = 'false', " +
 */